package bigo;

import bigo.data.MatrixMeta;
import bigo.data.Vector;
import bigo.lib.BinaryMatrixInputFormat;
import bigo.lib.BinaryMatrixOutputFormat;
import bigo.lib.CachedMatrix;
import bigo.lib.Combiner;
import bigo.lib.InputMatrix;
import bigo.lib.OutputMatrix;
import bigo.lib.TextMatrixInputFormat;
import bigo.lib.TextMatrixOutputFormat;
import java.lang.reflect.Field;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileContext;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapred.LocalJobRunner;
import org.apache.hadoop.mapreduce.Cluster;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.Tool;

/**
 *
 * @author songliu
 */
public abstract class BigODriver extends Configured implements Tool {

    public BigODriver() {
    }

    public void initFields(Configuration conf) {
        conf.set("driver.class", this.getClass().getName());

        for (Field f : this.getClass().getFields()) {
            try {
                if (f.get(this) instanceof MatrixMeta) {
                    //store our meta variables with its varible name hashcode.
                    conf.set("" + f.getName().hashCode(), f.get(this).toString());
                }
            } catch (IllegalArgumentException ex) {
                Logger.getLogger(BigODriver.class.getName()).log(Level.SEVERE, null, ex);
            } catch (IllegalAccessException ex) {
                Logger.getLogger(BigODriver.class.getName()).log(Level.SEVERE, null, ex);
            }
        }

    }

    public static void loadFields(Configuration conf) {
        try {
            //load MatrixMetas defined in class as public static
            Class driverClass = Class.forName(conf.get("driver.class"));
            
            for (Field f : driverClass.getFields()) {
                if (f.getType().isAssignableFrom(MatrixMeta.class)) {
                    String str = conf.get("" + f.getName().hashCode());
                    f.set(null, new MatrixMeta(str));
                }
            }
        } catch (Exception ex) {
            Logger.getLogger(BigODriver.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    public abstract void init(Job job);

    public abstract void preRun(Job job);

    public int run(String[] args) throws Exception {
        initFields(getConf());
        Job job = Job.getInstance(new Cluster(getConf()), getConf());
        init(job);
        job.createSymlink();
        job.setJarByClass(BigODriver.class);

        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(Vector.Wrapper.class);
        LocalJobRunner.setLocalMaxRunningMaps(job, 2);

        Class driverClass = this.getClass();

        for (Class c : driverClass.getDeclaredClasses()) {
            if (c.isAnnotationPresent(bigo.lib.Mapper.class)) {
                job.setMapperClass(c);
            }
            if (c.isAnnotationPresent(Combiner.class)) {
                job.setCombinerClass(c);
            }
            if (c.isAnnotationPresent(bigo.lib.Reducer.class)) {
                job.setReducerClass(c);
            }
        }

        Class inputFormat = null, outputFormat = null;
        for (Field f : driverClass.getDeclaredFields()) {
            if (f.isAnnotationPresent(bigo.lib.InputFormat.class)) {
                inputFormat = (Class) f.get(null);
                job.setInputFormatClass(inputFormat);
            }
            if (f.isAnnotationPresent(bigo.lib.OutputFormat.class)) {
                outputFormat = (Class) f.get(null);
                job.setOutputFormatClass(outputFormat);
            }
        }

        for (Field f : driverClass.getFields()) {
            if (f.isAnnotationPresent(CachedMatrix.class)) {
                ((MatrixMeta) f.get(null)).addCacheMatrix(job,
                        f.getAnnotation(CachedMatrix.class).overwrite());
                job.getConfiguration().setBoolean("cache.binary",
                        f.getAnnotation(CachedMatrix.class).binary());
            }

            if (f.isAnnotationPresent(InputMatrix.class)) {
                MatrixMeta meta = ((MatrixMeta) f.get(null));
                if (inputFormat.toString().equals(
                        TextMatrixInputFormat.class.toString())) {
                    TextMatrixInputFormat.addInputMatrix(job, meta);
                } else if (inputFormat.toString().equals(
                        BinaryMatrixInputFormat.class.toString())) {
                    BinaryMatrixInputFormat.addInputMatrix(job, meta);
                }
            }

            if (f.isAnnotationPresent(OutputMatrix.class)) {
                MatrixMeta meta =
                        ((MatrixMeta) f.get(null));
                if (outputFormat.toString().equals(
                        TextMatrixOutputFormat.class.toString())) {
                    TextMatrixOutputFormat.setOutputMatrix(job, meta);
                } else if (outputFormat.toString().equals(
                        BinaryMatrixOutputFormat.class.toString())) {
                    BinaryMatrixOutputFormat.setOutputMatrix(job, meta);
                }

                FileContext.getFileContext(getConf()).delete(
                        new Path(meta.path), true);

            }
        }

        preRun(job);
        return job.waitForCompletion(true) ? 0 : 1;
    }
}
